package com.shujia.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object Demo02WordCount {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo02WordCount")
      .master("local")
      .config("spark.sql.shuffle.partitions", 2) // 默认200
      .getOrCreate()

    // 统计每个单词的数量

    // 读取单词数据并构建DataFrame
    val lineDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "|")
      .schema("line String")
      .load("Spark/data/wordCnt/input")

    //    lineDF.show()

    // 将DataFrame注册成表 即可使用Spark SQL进行查询
    lineDF.createOrReplaceTempView("word_count")

    // SQL 的方式
    spark.sql(
      """
        |SELECT  t1.word
        |        ,count(*) as cnt
        |from (
        |    select  explode(split(line,",")) as word
        |    from word_count
        |) t1 group by t1.word
        |""".stripMargin).show()

    // 导入Spark SQL提供的所有的方法
    import org.apache.spark.sql.functions._
    // 导入spark sql的隐式转换
    import spark.implicits._

    // DSL 的方式
    lineDF
      .select(explode(split($"line", ",")) as "word")
      .groupBy($"word")
      .agg(count("*") as "cnt")
      .show()

    // RDD的方式
    val lineRDD: RDD[Row] = lineDF.rdd

    lineRDD
      // Row 表示DF中的一行数据 是具有结构的 可以通过getAs方法提取每一列数据
      .flatMap(row => {
        val line: String = row.getAs[String]("line")
        line.split(",")
      }).map(word => (word, 1))
      .reduceByKey(_ + _)
      .foreach(println)


    while (true) {

    }
  }

}
